
#include "zmq_message_handle.h"
#include "packagefilter.h"
#include "processhub.h"

#define CPIDREQUEST_TIMEOUT 350 // 5 min

ZmqMessageHandle::ZmqMessageHandle()
    : m_connectSize(0)
{
    std::string type_name = ProcessRouter::RPCRequest::descriptor()->full_name();
    rooftop_add_event(type_name.c_str()); 
    Rooftop::EventManager::getInstance()->addEventCallback(type_name.c_str(),
            boost::bind(&ZmqMessageHandle::onRequest, this, _1));

    type_name = ProcessRouter::RPCAnswer::descriptor()->full_name();
    rooftop_add_event(type_name.c_str()); 
    Rooftop::EventManager::getInstance()->addEventCallback(type_name.c_str(),
            boost::bind(&ZmqMessageHandle::onAnwser, this, _1));

    type_name = ProcessRouter::CommandGetFreePid::descriptor()->full_name();
    rooftop_add_event(type_name.c_str()); 
    Rooftop::EventManager::getInstance()->addEventCallback(type_name.c_str(),
            boost::bind(&ZmqMessageHandle::onGetFreePid, this, _1));


    type_name = ProcessRouter::CommandReleasePid::descriptor()->full_name();
    rooftop_add_event(type_name.c_str()); 
    Rooftop::EventManager::getInstance()->addEventCallback(type_name.c_str(),
            boost::bind(&ZmqMessageHandle::onReleasePid, this, _1));

    rooftop_add_event(CHILDPROCESSERROREVENT); 
    Rooftop::EventManager::getInstance()->addEventCallback(CHILDPROCESSERROREVENT,
            boost::bind(&ZmqMessageHandle::onChildProcessError, this, _1));
}

ZmqMessageHandle::~ZmqMessageHandle()
{
}

void ZmqMessageHandle::didRecevied()
{
    bool end = false;
    while (!end) {
        CMessage msg = channel()->recv();
        if (!msg.isNull()) {
            CMessageEvent event(msg, CMessageEvent::ZMQTYPE);
            Rooftop::EventManager::getInstance()->dispachEvent(event);
        } else
            end = true;
    }
}

void ZmqMessageHandle::didConnected()
{
}

void ZmqMessageHandle::didError()
{
}

bool ZmqMessageHandle::onRequest(Rooftop::EventBase &e)
{
    PackageEvent *event = static_cast<PackageEvent*>(&e);
    if (!event->package().has_request())
        return false;
    RPCRequest *request = event->package().mutable_request();
    
    CMessage message = event->message();

    request->set_clientid(request->clientid() + message.addressString());

    //std::cout << __func__ << " | " << request->aimpid() << " | " << request->clientid() << " | " << m_addressMap.size() << std::endl;

    ProcessRouter::FifoChannel* pchannel = ProcessRouter::ProcessHub::instance()->getChannel(request->aimpid());

    m_caimpidTimeMap[request->aimpid()] = time(NULL);

    if (!pchannel) {
        ProcessRouter::RPCAnswer answer;
        answer.set_clientid(request->clientid());
        answer.set_retval("Process not exists!");
        answer.set_exception(true);

        std::string ret;
        answer.SerializeToString(&ret);
        message.setContent(ret);
        channel()->send(message);
        return false;
    }

    std::string newcontent;
    event->package().SerializeToString(&newcontent);
    message.setContent(newcontent);

    int wret = pchannel->send(message);

    if (wret == -1) {
        perror("Pipe write error");
        ProcessRouter::RPCAnswer answer;
        answer.set_clientid(request->clientid());
        answer.set_retval("Pipe write error!");
        answer.set_exception(true);

        std::string ret;
        answer.SerializeToString(&ret);
        message.setContent(ret);
        channel()->send(message);
        return false;
    }

    m_addressMap.insert(std::make_pair(request->clientid(), message));
    m_cpidMap.insert(std::make_pair(request->clientid(), request->aimpid()));
    m_connectSize ++;
}

bool ZmqMessageHandle::onAnwser(Rooftop::EventBase &e)
{
    PackageEvent *event = static_cast<PackageEvent*>(&e);
    if (!event->package().has_answer())
        return false;
    RPCAnswer answer = event->package().answer();
    //std::cout << __func__ << " | " << answer.clientid() << " | " << m_addressMap.size() << std::endl;

    std::map<std::string, CMessage>::iterator it = m_addressMap.find(answer.clientid());
    std::map<std::string, int>::iterator cit = m_cpidMap.find(answer.clientid());

    if (cit != m_cpidMap.end()) {
        m_caimpidTimeMap[cit->second] = time(NULL);
        m_cpidMap.erase(cit);
    }

    if (it == m_addressMap.end())
        return false;

    std::string ret;
    answer.SerializeToString(&ret);
    CMessage message = it->second;
    message.setContent(ret);
    channel()->send(message);

    m_addressMap.erase(it);
    m_connectSize --;
}

bool ZmqMessageHandle::onChildProcessError(Rooftop::EventBase &e)
{
    ChildProceeErrorEvent *event = static_cast<ChildProceeErrorEvent*>(&e);
    int cpid = event->cpid();

    std::map<std::string, int>::iterator cit = m_cpidMap.begin();

    for (; cit != m_cpidMap.end(); cit ++) {
        if (cit->second == cpid) {
            std::cout << "child proccess error : " << cpid << std::endl;
            std::map<std::string, CMessage>::iterator it = m_addressMap.find(cit->first);

            if (it == m_addressMap.end())
                continue;

            std::string ret;
            RPCAnswer answer;
            answer.set_clientid(it->first);
            answer.set_retval("");
            answer.set_exception(true);
            answer.SerializeToString(&ret);
            CMessage message = it->second;
            message.setContent(ret);
            channel()->send(message);

            m_addressMap.erase(it);
            m_cpidMap.erase(cit);
            m_connectSize --;
        }
    }
}

bool ZmqMessageHandle::onGetFreePid(Rooftop::EventBase &e)
{
    PackageEvent *event = static_cast<PackageEvent*>(&e);
    if (!event->package().has_commandgetfreepid())
        return false;

    CMessage message = event->message();

    std::string ret;
    CommandGetFreePid command = event->package().commandgetfreepid();

    int pid = ProcessHub::instance()->info()->getFreePid(command.isstatic());

    if (pid == -1)
        printf("get free %d process pid : %d\n",command.isstatic(), pid);
    else {
        m_caimpidMap.insert(std::make_pair(command.clientid() + message.addressString(), pid));

        m_caimpidTimeMap.insert(std::make_pair(pid, time(NULL)));
    }

    command.set_freepid(pid);
    command.SerializeToString(&ret);

    message.setContent(ret);
    channel()->send(message);
}

bool ZmqMessageHandle::onReleasePid(Rooftop::EventBase &e)
{
    PackageEvent *event = static_cast<PackageEvent*>(&e);
    if (!event->package().has_commandreleasepid())
        return false;
    CMessage message = event->message();

    CommandReleasePid command = event->package().commandreleasepid();
    //std::cout << __func__ << " | " << command.freepid() << " | " << command.clientid() << std::endl;

    std::map<std::string, int>::iterator cit = m_caimpidMap.find(command.clientid() + message.addressString());

    int cpid = -1;
    if (cit != m_caimpidMap.end()) {
        cpid = cit->second;
        m_caimpidMap.erase(cit);
    }

    channel()->send(message);

    //如果已经没有客户端调用process,清除process计时.
    if (cpid == -1)
        return false;

    for (cit = m_caimpidMap.begin(); cit != m_caimpidMap.end(); cit ++)
        if (cit->second == cpid)
            return false;

    std::map<int, time_t>::iterator tit = m_caimpidTimeMap.find(cpid);
    if (tit != m_caimpidTimeMap.end())
        m_caimpidTimeMap.erase(tit);
}

/*
 * 这个函数在proccesshub类判断进程已经准备好退出或者意外退出之后调用.
 * 如果一个进程超过５分钟没有被调用,则认为这个进程已经被free了,清除进
 * 程状态并发送进程ERROR事件.
 */
bool ZmqMessageHandle::hasCpidRequest(int cpid)
{

    std::map<int, time_t>::iterator tit = m_caimpidTimeMap.find(cpid);
    if (tit != m_caimpidTimeMap.end()) {
        time_t ltime = tit->second;

        if (ltime <= 0) {
            m_caimpidTimeMap[cpid] = time(NULL);
            return false;
        }

        time_t now = time(NULL);
        if (now <= 0)
            return false;

        if ((now - ltime) > CPIDREQUEST_TIMEOUT) {

            m_caimpidTimeMap.erase(tit);
            ProcessHub::instance()->info()->removeCoreStatus(cpid);

            std::map<std::string, int>::iterator cit;

            for (cit = m_caimpidMap.begin(); cit != m_caimpidMap.end(); cit ++)
                if (cit->second == cpid)
                    m_caimpidMap.erase(cit);

            return false;
        } else
            return true;
    }

    return false;
}

void ZmqMessageHandle::doExit()
{
    std::map<std::string, CMessage>::iterator it = m_addressMap.begin();
    for (; it != m_addressMap.end(); it++ ) {
            std::string ret;
            RPCAnswer answer;
            answer.set_clientid(it->first);
            answer.set_retval("");
            answer.set_exception(true);
            answer.SerializeToString(&ret);
            CMessage message = it->second;
            message.setContent(ret);
            channel()->send(message);
            usleep(100000);
    }
}

int ZmqMessageHandle::connectSize()
{
    return m_connectSize;
}
